package com.dk.mq;

import com.dk.service.GoodsSevice;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.apache.rocketmq.spring.core.RocketMQPushConsumerLifecycleListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.ArrayList;
import java.util.List;


@Component
@RocketMQMessageListener(topic = "goodsId",consumerGroup = "dt")
public class FreemarkerRocketMq implements RocketMQListener<String> , RocketMQPushConsumerLifecycleListener {

    @Autowired
    private GoodsSevice goodsSevice;

    @Override
    public void onMessage(String s) {
    }

    @Override
    public void prepareStart(DefaultMQPushConsumer defaultMQPushConsumer) {
        defaultMQPushConsumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list,
                                                            ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                MessageExt messageExt = list.get(0);
                messageExt.getTopic();//获取主题
                messageExt.getMsgId();//获取id
                messageExt.setReconsumeTimes(3);//设置重试次数3次
                String str = new String(messageExt.getBody());//获取消息
//                String[] split = str.split(",");
//                List<Long> longs = new ArrayList<>();
//                for (int i = 0; i < split.length; i++) {
//                    longs.add(Long.valueOf(split[i]));
//                }
//                System.out.println(longs.size());
//                for (int i = 0; i < longs.size(); i++) {
//                    goodsSevice.creagePage(longs.get(i));
                    goodsSevice.creagePage(Long.valueOf(str));
//                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
    }
}
